package com.lagou.bak;

import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

public class BroadCastStateRun {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        Actition ac1 = new Actition(1001l, "login");
        Actition ac2 = new Actition(1003l, "pay");
        Actition ac3 = new Actition(1002l, "car");
        Actition ac4 = new Actition(1001l, "logout");
        Actition ac5 = new Actition(1003l, "car");
        Actition ac6 = new Actition(1002l, "login");
        DataStreamSource<Actition> actions = env.fromElements(ac1, ac2, ac3, ac4, ac5, ac6);

        PatternBean p1 = new PatternBean("car", "logout");
        PatternBean p2 = new PatternBean("login", "logout");
        DataStreamSource<PatternBean> patterns = env.fromElements(p2);

        //根据key进行分组，确保同一个用户的操作行为在同一个Task中处理
        KeyedStream<Actition, Long> actionByUser = actions.keyBy(action -> action.getUserId());
//        actionByUser.print();

        //准备broadcast state
        MapStateDescriptor<Void, PatternBean> bcStateDescriptor = new MapStateDescriptor<>("patterns", Types.VOID, Types.POJO(PatternBean.class));
        BroadcastStream<PatternBean> bcedPatterns = patterns.broadcast(bcStateDescriptor);
        SingleOutputStreamOperator<Tuple2<Long, PatternBean>> process = actionByUser.connect(bcedPatterns).process(new PatternEvaluator());
        process.print();
        env.execute();
    }

    public static class PatternEvaluator extends KeyedBroadcastProcessFunction<Long, Actition, PatternBean, Tuple2<Long, PatternBean>> {
        //
        ValueState<String> prevActionState;

        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println(Thread.currentThread().getName() + "....open...");
            prevActionState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastAction", Types.STRING));
//            System.out.println(prevActionState);
        }

        @Override
        public void processElement(Actition value, ReadOnlyContext ctx, Collector<Tuple2<Long, PatternBean>> out) throws Exception {
            System.out.println(Thread.currentThread().getName() + "....value:" + value);
            PatternBean pattern = ctx.getBroadcastState(new MapStateDescriptor<>("patterns", Types.VOID, Types.POJO(PatternBean.class))).get(null);
//            System.out.println("....processElement...pattern:" + pattern);
            String prevAction = prevActionState.value();
//            System.out.println(Thread.currentThread().getName() + "....processElement...prevAction:" + prevAction);
            if (pattern != null && prevAction != null) {
                if (pattern.getFirstAction().equals(prevAction) && pattern.getSecondAction().equals(value.getAction())) {
//                    System.out.println("......................out.collect...");
                    out.collect(new Tuple2<>(ctx.getCurrentKey(), pattern));
                }
            }
            prevActionState.update(value.getAction());

        }

        @Override
        public void processBroadcastElement(PatternBean value, Context ctx, Collector<Tuple2<Long, PatternBean>> out) throws Exception {
//            System.out.println("...processBroadcastElement...");
//            System.out.println(Thread.currentThread().getName() + "....value:" + value);
            BroadcastState<Void, PatternBean> bcstate = ctx.getBroadcastState(new MapStateDescriptor<>("patterns", Types.VOID, Types.POJO(PatternBean.class)));
            bcstate.put(null, value);

        }
    }
}





















